package com.niit.spark.sparkSql


import com.google.gson.JsonObject
import com.niit.utils.MyKafkaUtils
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.parquet.format.LogicalType.JSON
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import java.util.Properties



/**
 * @author 杨铭
 *         2022/11/19,10:11
 */
object sparkSql {
  def main(args: Array[String]): Unit = {

    System.setProperty("hadoop.home.dir", "D:\\ScalaProject\\hadoop-2.7.3")
    System.setProperty("HADOOP_USER_NAME", "root")
    val conf= new SparkConf().setMaster("local[*]").setAppName("Spark5")
    val sc= new SparkContext(conf)

    val sqlCon = new SQLContext(sc)

    val df=sqlCon.read.json("D:\\order.json")
    df.createOrReplaceTempView("order")

    val Yes=sqlCon.sql("select  orderCategory as category,isCorrect as is_correct,count(*) as num from order where isCorrect='Y' group by orderCategory,isCorrect")
    val No=sqlCon.sql("select  orderCategory as category,isCorrect as is_correct,count(*) as num from order where isCorrect='N' group by orderCategory,isCorrect")

    val prop=new Properties()
    prop.put("user","root")
    prop.put("password","hyt297375")
    Yes.write.mode("append").jdbc("jdbc:mysql://localhost:3306/niitspark?serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=UTF-8","allcategory",prop)
    No.write.mode("append").jdbc("jdbc:mysql://localhost:3306/niitspark?serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=UTF-8","allcategory",prop)
    /* Yes.rdd.foreach(row=>{
       val number:Long = row.getAs[Long]("number")
      val orderCategory:String=row.getAs[String]("orderCategory")
       val json =new JsonObject()
       json.addProperty("orderCategory",orderCategory)
       json.addProperty("number",number)
       json.addProperty("isValid","Y")


      if(json.toString!=null){
        MyKafkaUtils.send("task5",json.toString)
      }

     })*/



    /*No.rdd.foreach(row=>{
      val number:Long = row.getAs[Long]("number")
      val orderCategory:String=row.getAs[String]("orderCategory")
      val json2 =new JsonObject()
      json2.addProperty("orderCategory",orderCategory)
      json2.addProperty("number",number)
      json2.addProperty("isValid","N")

      if(json2.toString!=null){

        MyKafkaUtils.send("task5",json2.toString)
      }

    })*/

  }
}
